package com.grow.greenhouse.utils;

import com.grow.greenhouse.common.WebSocket;
import com.grow.greenhouse.job.MessageListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

public class MqttUtil {

    private MqttClient client;
    private MqttConnectOptions options;
    @Resource
    private WebSocket webSocket;
    public MqttUtil(String brokerAddress, int brokerPort, String clientId, String userName, String password) {
        try {
            client = new MqttClient("tcp://" + brokerAddress + ":" + brokerPort, clientId);
            options = new MqttConnectOptions();
            options.setAutomaticReconnect(true);
            options.setCleanSession(true);
            options.setConnectionTimeout(10);
            options.setUserName(userName);
            options.setPassword(password.toCharArray());

            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    // 这里可以添加连接丢失时的处理逻辑
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // 这里可以添加接收到消息时的处理逻辑
                    System.out.println("Received message: " + new String(message.getPayload()) + "\nOn topic: " + topic);
                    notifyListeners(topic, message);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // 这里可以添加消息发送完成时的处理逻辑
                }
            });

            client.connect(options);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void subscribe(String topic, int qos) {
        try {
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
    private List<MessageListener> listeners = new ArrayList<>();

    public void addMessageListener(MessageListener listener) {
        listeners.add(listener);
    }

    private void notifyListeners(String topic, MqttMessage message) {
        for (MessageListener listener : listeners) {
            listener.onMessageReceived(topic, message);
        }
    }
    public void publish(String topic, String payload, int qos) {
        try {
            MqttMessage message = new MqttMessage();
            message.setPayload(payload.getBytes());
            message.setQos(qos);
            client.publish(topic, message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    // 断开连接和清理资源
    public void disconnect() {
        try {
            if (client != null) {
                client.disconnect();
                client.close();
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}
